package com.yahoo.ycsb.db;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Properties;
import java.util.Random;
import java.util.Vector;

public class Test {

    /**
     * @param args
     */
    public static void main(String[] args) {
         if (args.length != 3) {
             System.out.println(Integer.MAX_VALUE);
             
            System.out.println("Please specify a threadcount, columnfamily and operation count");
            System.exit(0);
        }

        final int keyspace = 10000; // 120000000;

        final int threadcount = Integer.parseInt(args[0]);

        final String columnfamily = args[1];

        final int opcount = Integer.parseInt(args[2]) / threadcount;

        Vector<Thread> allthreads = new Vector<Thread>();

        for (int i = 0; i < threadcount; i++) {
            Thread t = new Thread() {

                public void run() {
                    try {
                        Random random = new Random();

                        HBaseClient cli = new HBaseClient();

                        Properties props = new Properties();
                        props.setProperty("columnfamily", columnfamily);
                        props.setProperty("debug", "true");
                        cli.setProperties(props);

                        cli.init();

                        // HashMap<String,String> result=new HashMap<String,String>();

                        long accum = 0;

                        for (int i = 0; i < opcount; i++) {
                            int keynum = random.nextInt(keyspace);
                            String key = "user" + keynum;
                            long st = System.currentTimeMillis();
                            int rescode;
                            /*
                             * HashMap hm = new HashMap(); 
                             * hm.put("field1","value1"); 
                             * hm.put("field2","value2");
                             * hm.put("field3","value3"); 
                             * rescode=cli.insert("table1",key,hm);
                             * HashSet<String> s = new HashSet(); 
                             * s.add("field1"); 
                             * s.add("field2"); 
                             * rescode=cli.read("table1", key, s, result);
                             * 
                             * rescode=cli.delete("table1",key); 
                             * rescode=cli.read("table1", key, s, result);
                             */
                            HashSet<String> scanFields = new HashSet<String>();
                            scanFields.add("field1");
                            scanFields.add("field3");
                            Vector<HashMap<String, String>> scanResults = new Vector<HashMap<String, String>>();
                            rescode = cli.scan("table1", "user2", 20, null, scanResults);

                            long en = System.currentTimeMillis();

                            accum += (en - st);

                            if (rescode != HBaseClient.Ok) {
                                System.out.println("Error " + rescode + " for " + key);
                            }

                            if (i % 1 == 0) {
                                System.out.println(i + " operations, average latency: "
                                                   + (((double) accum) / ((double) i)));
                            }
                        }

                         System.out.println("Average latency: "+(((double)accum)/((double)opcount)));
                        //System.out.println("Average get latency: "+(((double)cli.TotalGetTime)/((double)cli.TotalGetOps)));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            allthreads.add(t);
        }

        long st = System.currentTimeMillis();
        for (Thread t : allthreads) {
            t.start();
        }

        for (Thread t : allthreads) {
            try {
                t.join();
            } catch (InterruptedException e) {
            }
        }
        long en = System.currentTimeMillis();

        System.out.println("Throughput: " + ((1000.0) * (((double) (opcount * threadcount)) / ((double) (en - st))))
                           + " ops/sec");

    }
}
